package com.tuanzhang.zhw.rpc.client;


import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 连接管理器:
 * 地址解析
 * 建立连接
 * 连接缓存
 * 失败监听(重试，移除)
 */
@Slf4j
public class RpcConnectManager {

    public RpcConnectManager() {

    }

    /**
     * 用于发起异步连接的线程池
     */
    private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10240));

    /**
     * 存放所有服务的链接
     */
    private Map<InetSocketAddress, RpcClientHandler> connectedHandlerMap = new ConcurrentHashMap<InetSocketAddress, RpcClientHandler>();

    /**
     * 已连接成功的RpcClientHandler
     */
    private CopyOnWriteArrayList<RpcClientHandler> connectedHandlerList = new CopyOnWriteArrayList<RpcClientHandler>();

    /**
     * connectedCondition#signalAll唤醒另外一端的线程(阻塞的状态中): 告知有新连接加入
     */
    private ReentrantLock connectedLock = new ReentrantLock();

    private Condition connectedCondition = connectedLock.newCondition();
    private long connectTimeoutMills = 6000;// 连接选择器等待的超时时间
    private volatile boolean isRunning = true;// 程序开关: 连接管理器运行状态
    private volatile AtomicInteger handlerIndex = new AtomicInteger(0);// 当前连接选择器所选的业务处理器索引 => volatile是多余的, 写不写都可以, 因为AtomicInteger的value本身就是用了volatile修饰


    /**
     * 连接线程组  默认cpu*2
     */
    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup();


    /**
     * 发起连接
     *
     * @param serverAddress
     */
    public void connect(final String serverAddress) {
        List<String> allServerAddress = Arrays.asList(serverAddress.split(","));
        updateConnectedServer(allServerAddress);

    }

    /**
     * 	add connect List<String> serverAddress
     * @param serverAddress
     */
    public void connect(List<String> serverAddress) {
        updateConnectedServer(serverAddress);
    }

    /**
     * 选择连接处理器
     *
     * @return
     */
    public RpcClientHandler chooseHandler() {
        //防止并发问题 copy一份数据
        CopyOnWriteArrayList<RpcClientHandler> handlers = (CopyOnWriteArrayList<RpcClientHandler>) this.connectedHandlerList.clone();
        int size = handlers.size();
        while (isRunning && size <= 0) {
            try {
                if (waitingAvailableHandler()) { // 等待 6s
                    handlers = (CopyOnWriteArrayList<RpcClientHandler>) this.connectedHandlerList.clone();
                    size = handlers.size();
                }
            } catch (InterruptedException e) {
                log.error("服务中断");
                throw new RuntimeException("服务中断", e);
            }

        }

        if (!isRunning) {
            return null;
        }
        //取模轮训选择业务处理器
        return handlers.get((handlerIndex.getAndAdd(1) + size) % size);

    }

    /**
     * 关闭连接管理器服务
     */
    public void stop() {
        isRunning = false;
        // 关闭handler连接资源
        RpcClientHandler[] rpcClientHandlers = connectedHandlerList.toArray(new RpcClientHandler[0]);
        for (RpcClientHandler rpcClientHandler : rpcClientHandlers) {
            clearConnected((InetSocketAddress) rpcClientHandler.getRemotePeer());
        }

        // 唤醒所有正在阻塞的线程, 因为程序开关发生了变化 => 这次唤醒会使得所有线程统统选择了空的业务处理器
        // 或者可以让阻塞的线程进行等待超时, 然后抛出了异常就终止了 => 但是这样不优雅
        signalAvailableHandler();

        // 关闭资源
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }

    /**
     * 重连
     *
     * @param handler
     * @param socketAddress
     */
    public void reconnect(RpcClientHandler handler, SocketAddress socketAddress) {
        if (handler != null) {
            handler.close();

            connectedHandlerList.remove(handler);
            connectedHandlerMap.remove(socketAddress);
        }
        //  异步重连
        connectAsync((InetSocketAddress) socketAddress);

    }


    /**
     * 更新连接
     *
     * @param allServerAddress {192.168.1.100:8765,192.168.1.101:8765}
     */
    public void updateConnectedServer(List<String> allServerAddress) {
        if (CollectionUtils.isEmpty(allServerAddress)) {
            log.error("没有服务地址");
            //清除 缓存
            clearAllConnected();
            //  todo
            return;
        }

        // 1、 解析地址
        Set<InetSocketAddress> newAllServerNodeSet = new HashSet<InetSocketAddress>();
        allServerAddress.forEach(serverAddress -> {
            String[] array = serverAddress.split(":");
            if (array.length == 2) {
                String ipHost = array[0];
                int port = Integer.parseInt(array[1]);
                InetSocketAddress inetSocketAddress = new InetSocketAddress(ipHost, port);
                newAllServerNodeSet.add(inetSocketAddress);
            }
        });

        // 2、建立连接
        Set<InetSocketAddress> existedInetSocketAddresses = connectedHandlerMap.keySet();
        newAllServerNodeSet.forEach(serverAddress -> {
            if (!existedInetSocketAddresses.contains(serverAddress)) {
                // 如果有新的ip 进行连接
                connectAsync(serverAddress);
            }
        });

        //  3、如果 allServerAddress没出现对应的地址, 则需要从缓存中移除
        RpcClientHandler[] rpcClientHandlers = connectedHandlerList.toArray(new RpcClientHandler[0]);
        for (int i = 0; i < rpcClientHandlers.length; i++) {
            RpcClientHandler rpcClientHandler = rpcClientHandlers[i];
            SocketAddress remotePeer = rpcClientHandler.getRemotePeer();
            if (newAllServerNodeSet.contains(remotePeer)) {
                log.info("删除服务 " + remotePeer);
                clearConnected((InetSocketAddress) remotePeer);
            }

        }

    }


    /**
     * 异步连接
     *
     * @param serverAddress
     */
    private void connectAsync(InetSocketAddress serverAddress) {
        threadPoolExecutor.submit(() -> {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    // TCP默认开启了 Nagle 算法，该算法的作用是尽可能的发送大数据快，减少网络传输。TCP_NODELAY 参数的作用就是控制是否启用 Nagle 算法。
                    .option(ChannelOption.TCP_NODELAY, Boolean.TRUE)
                    .handler(new RpcClientInitializer());
            connect(bootstrap, serverAddress);
        });
    }

    private void connect(Bootstrap bootstrap, InetSocketAddress remotePeer) {
        // 1、建立连接
        ChannelFuture channelFuture = bootstrap.connect(remotePeer);

        // 2、连接失败监听器
        channelFuture.channel().closeFuture().addListener((ChannelFutureListener) future -> {
            log.info("rpc连接失败的地址,serverAddress={}", remotePeer);
            future.channel().eventLoop().schedule(() -> {
                clearConnected(remotePeer);
                //  ToDo 不能一直重试
                connect(bootstrap, remotePeer);
            }, 3, TimeUnit.SECONDS);
        });

        // 3、连接成功添加监听
        channelFuture.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.info("连接成功, serverAddress={}", remotePeer);
                RpcClientHandler handler = future.channel().pipeline().get(RpcClientHandler.class);
                addHandler(handler);
            }
        });
    }

    private void addHandler(RpcClientHandler handler) {
        connectedHandlerList.add(handler);
        connectedHandlerMap.put((InetSocketAddress) handler.getChannel().remoteAddress(), handler);
        // 唤醒可用的业务执行器 signalAvailableHandler
        signalAvailableHandler();
    }

    /**
     * 唤醒另外一端的线程(阻塞的状态中): 告知连接选择处理器有新连接加入
     */
    private void signalAvailableHandler() {
        connectedLock.lock();
        try {
            connectedCondition.signalAll();
        } finally {
            connectedLock.unlock();
        }
    }

    /**
     * 等待新连接接入
     */
    private boolean waitingAvailableHandler() throws InterruptedException {
        connectedLock.lock();
        try {
            return connectedCondition.await(this.connectTimeoutMills, TimeUnit.MILLISECONDS);
        } finally {
            connectedLock.unlock();
        }
    }


    /**
     * 连接失败时, 及时清除资源, 清空缓存: 单独清理失败的
     */
    private void clearConnected(InetSocketAddress remotePeer) {
        // 然后从connectedHandlerMap缓存中移除指定的rpcClientHandler
        RpcClientHandler handler = connectedHandlerMap.get(remotePeer);
        if (handler != null) {
            handler.close();
            connectedHandlerMap.remove(remotePeer);
        }
        connectedHandlerList.remove(handler);
    }

    /**
     * 清除所有缓存
     */
    private void clearAllConnected() {
        for (RpcClientHandler rpcClientHandler : connectedHandlerList) {
            // 通过rpcClientHandler找到remotePeer => 在rpcClientHandler缓存了remotePeer的引用
            SocketAddress remotePeer = rpcClientHandler.getRemotePeer();

            // 然后从connectedHandlerMap缓存中移除指定的rpcClientHandler
            RpcClientHandler handler = connectedHandlerMap.get(remotePeer);
            if (handler != null) {
                handler.close();
                connectedHandlerMap.remove(remotePeer);
            }
        }
        connectedHandlerList.clear();
    }

}
